2025年11月01日

by smallnest

langchain + MCP：如虎添翼

MCP技术毋须多言了，上半年火的一塌糊涂，现在进入冷静期了。

langchain 本身就很方便的集成进程内的工具，但是加上 MCP的功能，就如虎添翼，可以充分利用网上上万的MCP的服务。

langchain 自从上个月融资了1.25亿美元之后，资金充足，也更加有动力推进产品的演化，相继发布了langchain/langgraph 1.0的版本。 langchain 1.0中统一了agent的创建，使用create_agent代替之前的create_tool_calling_agent、create_react_agent、create_json_agent、create_xml_agent等。

这篇文章介绍 2（MCP的两种模式sse、stdio） x 2 (经典的langchain agent和1.0最新版create_agent两个模式)一共4个例子，介绍了langchain如何使用MCP 工具丰富其功能。

先决条件

在运行这些示例之前，请确保您已安装必要的库。您通常可以使用 pip 进行安装：

1	pip install langchain langchain-openai langchain-classic langchain-mcp-adapters mcp

此外，请确保您的环境中已安装 python3。

共同组件

这两个示例都共享几个核心组件和一个共同目标：使用 LangChain Agent 通过 MCP 服务器提供的工具来回答一个简单的数学问题。

LLM 配置：两个示例都使用 ChatOpenAI，配置为 deepseek-v3 模型，温度为 0。
这是我使用百度云上提供的deepseek服务，你要是使用deepseek官方的服务，需要修改模型为deepseek-chat。
我已经把KEY和调用地址配置在环境变量中了,所以在代码中不用显示指定：

1 2	export OPENAI_API_KEY=bce-v3/abcsfsfdskgergerthntjrweeuidfu8324refbif3 export OPENAI_API_BASE=https://qianfan.baidubce.com/v2

创建模型对象：

1	llm = ChatOpenAI(model="deepseek-v3", temperature=0)

Agent 提示：ChatPromptTemplate 用于定义 Agent 的角色并构建对话。它包括一个系统消息、可选的聊天历史记录、人类输入以及 Agent 暂存区（用于规划其行动）的占位符。

prompt = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一个可以使用工具的得力助手。 தயவுசெய்து கருவிகளைப் பயன்படுத்தவும்."),
        MessagesPlaceholder("chat_history", optional=True),
        ("human", "{input}"),
        MessagesPlaceholder("agent_scratchpad"),
    ]
)

Agent 创建和执行：create_tool_calling_agent 函数用于构建能够使用工具的 Agent ，AgentExecutor 用于运行 Agent 。

1 2	agent = create_tool_calling_agent(llm, tools, prompt) agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) # SSE 示例中 verbose=False

任务： Agent 被调用，输入为“123 + 456 等于多少？”。

MCP 服务

Stdio服务

import asyncio
from mcp.server import FastMCP
# 创建一个服务器实例
server = FastMCP(name="math_server", log_level="ERROR")
# 定义并注册 add 工具
@server.tool()
def add(a: int, b: int) -> int:
    """将两个整数相加。"""
    return a + b
async def main():
    await server.run_stdio_async()
if __name__ == "__main__":
    asyncio.run(main())

sse 服务

import asyncio
from typing import Annotated
from mcp.server import FastMCP
# 创建一个服务器实例
server = FastMCP(name="math_server", instructions="一个可以做加法的简单数学服务器。", log_level="ERROR")
# 定义并注册 add 工具
@server.tool()
def add(
    a: Annotated[int, "第一个整数"],
    b: Annotated[int, "第二个整数"]
) -> int:
    """将两个整数相加。"""
    return a + b
async def main():
    await server.run_sse_async()
if __name__ == "__main__":
    asyncio.run(main())

示例 1：`example_1_mcp_tool_stdio.py`（标准 I/O 通信）

此示例演示了如何设置一个 MCP 服务器，该服务器通过标准输入和输出流与客户端通信。这适用于本地、单进程交互，其中服务器可以作为子进程生成。

目的

example_1_mcp_tool_stdio.py 脚本展示了如何：

定义一个通过 stdio 暴露工具的 MCP 服务器。
创建一个 stdio_client 来连接到此服务器。
将服务器提供的工具加载到 LangChain Agent 中。
使用 Agent 解决需要加载工具的问题。

关键组件和代码

import asyncio
import os
from pathlib import Path
from langchain_openai import ChatOpenAI
from langchain_classic.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_mcp_adapters.tools import load_mcp_tools
from mcp.client.stdio import stdio_client
from mcp import ClientSession, StdioServerParameters
async def run_mcp_tool_example():
    # ... (LLM 和提示设置，如共同组件中所述) ...
    # 1. 定义 MCP 数学服务器的路径
    # 此行构建了 stdio 数学服务器脚本的绝对路径。
    # `Path(__file__).parent` 获取当前脚本的目录。
    mcp_server_path = Path(__file__).parent / "mcp_math_server_stdio.py"
    # 2. 为 MCP 服务器设置 stdio 客户端参数
    # StdioServerParameters 指定如何运行 MCP 服务器。
    # `command="python3"` 指示解释器。
    # `args=[str(mcp_server_path)]` 提供要作为参数执行的脚本。
    server_params = StdioServerParameters(
        command="python3",
        args=[str(mcp_server_path)],
    )
    # 3. 建立 stdio 客户端会话
    # `stdio_client(server_params)` 创建一个异步上下文管理器，
    # 它生成服务器进程并提供用于通信的读/写流。
    async with stdio_client(server_params) as (read, write):
        # 4. 创建 MCP 客户端会话
        # `ClientSession` 通过提供的读/写流管理 MCP 协议。
        async with ClientSession(read, write) as session:
            # 5. 初始化连接
            # 此步骤对于客户端和服务器建立 MCP 连接至关重要。
            await session.initialize()
            # 6. 从 MCP 服务器获取工具
            # `load_mcp_tools(session)` 从 MCP 服务器获取工具定义
            # 并将它们转换为 LangChain 兼容的工具对象。
            tools = await load_mcp_tools(session)
            # ... ( Agent 创建和执行，如共同组件中所述) ...
            print("正在调用 agent 回答一个数学问题...")
            response = await agent_executor.ainvoke(
                {"input": "123 + 456 等于多少？", "chat_history": []}
            )
            print(f"Agent 回答: {response['output']}")
if __name__ == "__main__":
    asyncio.run(run_mcp_tool_example())

如何运行

要运行此示例，只需执行 Python 脚本：

1	python3 example_1_mcp_tool_stdio.py

您将看到 Agent 的思考过程和最终答案打印到控制台。

示例 2：`example_1_mcp_tool_sse.py`（服务器发送事件通信）

此示例演示了如何集成一个 MCP 服务器，该服务器通过 HTTP 上的服务器发送事件（SSE）暴露其工具。此方法更适用于工具服务器可能是独立网络服务的情况。

目的

example_1_mcp_tool_sse.py 脚本展示了如何：

将 MCP SSE 服务器作为单独的进程启动。
创建一个 MultiServerMCPClient 来连接到此基于 HTTP 的服务器。
将服务器提供的工具加载到 LangChain Agent 中。
使用 Agent 解决需要加载工具的问题。

关键组件和代码演练

import asyncio
import os
from pathlib import Path
import sys
from langchain_openai import ChatOpenAI
from langchain_classic.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_mcp_adapters.client import MultiServerMCPClient
async def run_mcp_tool_example():
    # ... (LLM 和提示设置，如共同组件中所述) ...
    # 1. 启动 MCP 数学服务器作为子进程
    # 这将 `mcp_math_server_sse.py` 脚本作为单独的 Python 进程启动。
    # `sys.executable` 确保使用正确的 Python 解释器。
    # `stdout=asyncio.subprocess.PIPE` 和 `stderr=asyncio.subprocess.PIPE`
    # 捕获子进程的输出，尽管此处未明确读取。
    mcp_server_path = Path(__file__).parent / "mcp_math_server_sse.py"
    server_process = await asyncio.create_subprocess_exec(
        sys.executable, str(mcp_server_path),
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    # 2. 等待服务器启动
    # 添加一个小的延迟，以使 SSE 服务器有时间初始化并开始监听。
    await asyncio.sleep(5)
    # 3. 为 MCP 服务器设置客户端
    # `MultiServerMCPClient` 用于基于 HTTP 的 MCP 服务器。
    # 它接受一个字典，其中键是服务器名称（例如，“math”），值
    # 指定传输类型（“sse”）和 SSE 端点的 URL。
    client = MultiServerMCPClient(
        {
            "math": {
                "transport": "sse",
                "url": "http://localhost:8000/sse",
            },
        }
    )
    # 4. 从 MCP 服务器获取工具
    # `client.get_tools()` 连接到指定的 SSE 端点，
    # 检索工具定义，并将其作为 LangChain 工具提供。
    tools = await client.get_tools()
    # ... ( Agent 创建和执行，如共同组件中所述) ...
    print("正在调用 agent 回答一个数学问题...")
    response = await agent_executor.ainvoke(
        {"input": "123 + 456 等于多少？", "chat_history": []}
    )
    print(f"Agent 回答: {response['output']}")
    # 5. 终止服务器进程
    # 通过终止已启动的子进程进行清理非常重要。
    server_process.terminate()
    await server_process.wait()
if __name__ == "__main__":
    asyncio.run(run_mcp_tool_example())

如何运行

要运行此示例，只需执行 Python 脚本：

1	python3 example_1_mcp_tool_sse.py

您将看到 Agent 的思考过程和最终答案打印到控制台。请注意，在此示例中，AgentExecutor 的 verbose 标志设置为 False，以避免 SSE 服务器子进程产生过多的输出。

比较：Stdio 与 SSE

Stdio（标准 I/O）：
- 简单性：更易于设置本地、单机通信。
- 执行：MCP 服务器通常作为客户端应用程序的子进程生成。
- 用例：适用于紧密耦合的组件，或者当您希望将工具服务器直接与应用程序捆绑时。
SSE（服务器发送事件）：
- 灵活性：允许 MCP 服务器作为独立的网络服务运行，可能在不同的机器上。
- 可伸缩性：可以成为更大微服务架构的一部分。
- 执行：客户端连接到正在运行的 HTTP 端点。服务器需要单独启动（如示例中通过 asyncio.create_subprocess_exec 所示）。
- 用例：适用于分布式系统、基于 Web 的应用程序，或者当工具服务器需要被多个客户端访问时。

这两种方法都有效地允许 LangChain Agent 发现和利用使用多客户端协议定义的工具，从而为这些工具的部署和访问提供了灵活性。

这两个示例都是langchain经典的Agent开发模型，他们演示了langchain将 MCP 工具与 LangChain Agent 集成的强大功能和灵活性。无论它们是通过 stdio 还是 SSE 暴露。这使得健壮且可伸缩的 Agent 应用程序成为可能。

接下来我介绍如何使用 Langchain 1.0 中新的 create_agent 函数来构建 Agent ，并将其与 MCP 工具集成。与之前的示例不同，create_agent 提供了一种更简洁的方式来定义 Agent ，直接接受 LLM 模型、工具列表和系统提示，并返回一个可流式传输的图（graph）对象。

示例 3：使用 Langchain 1.0 create_agent 的 MCP SSE 工具集成

此示例展示了如何将 MCP SSE 服务器提供的工具与 Langchain 1.0 的 create_agent 函数结合使用。它演示了如何启动 SSE 服务器、获取工具，然后使用新的 Agent 创建和调用模式来解决数学问题。

import asyncio
import os
from pathlib import Path
import sys
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
async def run_mcp_sse_new_agent_example():
    server_process = None
    try:
        # 启动 MCP 数学服务器作为子进程
        mcp_server_path = Path(__file__).parent / "mcp_math_server_sse.py"
        server_process = await asyncio.create_subprocess_exec(
            sys.executable, str(mcp_server_path),
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        # 等待服务器启动
        await asyncio.sleep(5)
        # 为 MCP 服务器设置客户端
        client = MultiServerMCPClient(
            {
                "math": {
                    "transport": "sse",
                    "url": "http://localhost:8000/sse",
                },
            }
        )
        # 从 MCP 服务器获取工具
        tools = await client.get_tools()
        # 使用用户偏好配置 LLM
        llm = ChatOpenAI(model="deepseek-v3", temperature=0)
        # 创建一个 agent
        graph = create_agent(
            model=llm,
            tools=tools,
            system_prompt="你是一个可以使用工具的得力助手。 தயவுசெய்து கருவிகளைப் பயன்படுத்தவும்.",
        )
        print("正在调用 agent 回答一个数学问题...")
        inputs = {"messages": [{"role": "user", "content": "123 + 456 等于多少？"}]}
        async for chunk in graph.astream(inputs, stream_mode="updates"):
            print(chunk)
    finally:
        if server_process and server_process.returncode is None:
            print("\nTerminating server process...")
            server_process.terminate()
            await server_process.wait()
        elif server_process and server_process.returncode is not None:
            print(f"\nServer process exited with code: {server_process.returncode}")
        if server_process and server_process.stderr:
            stderr_output = await server_process.stderr.read()
            if stderr_output:
                print("\nServer stderr output:")
                print(stderr_output.decode())
if __name__ == "__main__":
    asyncio.run(run_mcp_sse_new_agent_example())

示例 4：使用 Langchain 1.0 create_agent 的 MCP Stdio 工具集成

此示例演示了如何将 MCP Stdio 服务器提供的工具与 Langchain 1.0 的 create_agent 函数结合使用。它展示了如何通过 stdio 客户端连接到服务器、获取工具，然后使用新的 Agent 创建和调用模式来解决数学问题。

import asyncio
import os
from pathlib import Path
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain_mcp_adapters.tools import load_mcp_tools
from mcp.client.stdio import stdio_client
from mcp import ClientSession, StdioServerParameters
async def run_mcp_stdio_new_agent_example():
    # 定义 MCP 数学服务器的路径
    mcp_server_path = Path(__file__).parent / "mcp_math_server_stdio.py"
    # 为 MCP 服务器设置 stdio 客户端
    server_params = StdioServerParameters(
        command="python3",
        args=[str(mcp_server_path)],
    )
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化连接
            await session.initialize()
            # 从 MCP 服务器获取工具
            tools = await load_mcp_tools(session)
            # 使用用户偏好配置 LLM
            llm = ChatOpenAI(model="deepseek-v3", temperature=0)
            # 创建一个 agent
            graph = create_agent(
                model=llm,
                tools=tools,
                system_prompt="你是一个可以使用工具的得力助手。 தயவுசெய்து கருவிகளைப் பயன்படுத்தவும்.",
            )
            print("正在调用 agent 回答一个数学问题...")
            inputs = {"messages": [{"role": "user", "content": "123 + 456 等于多少？"}]}
            async for chunk in graph.astream(inputs, stream_mode="updates"):
                print(chunk)
if __name__ == "__main__":
    asyncio.run(run_mcp_stdio_new_agent_example())

鸟窝

《Go語言全功能開發養成書》繁体中文版发售。一书在手，并发无忧

langchain + MCP：如虎添翼

先决条件

共同组件

MCP 服务

Stdio服务

sse 服务

示例 1：`example_1_mcp_tool_stdio.py`（标准 I/O 通信）

目的

关键组件和代码

如何运行

示例 2：`example_1_mcp_tool_sse.py`（服务器发送事件通信）

目的

关键组件和代码演练

如何运行

比较：Stdio 与 SSE

示例 3：使用 Langchain 1.0 create_agent 的 MCP SSE 工具集成

示例 4：使用 Langchain 1.0 create_agent 的 MCP Stdio 工具集成

先决条件

共同组件

MCP 服务

Stdio服务

sse 服务

示例 1：example_1_mcp_tool_stdio.py（标准 I/O 通信）

目的

关键组件和代码

如何运行

示例 2：example_1_mcp_tool_sse.py（服务器发送事件通信）

目的

关键组件和代码演练

如何运行

比较：Stdio 与 SSE

示例 3：使用 Langchain 1.0 create_agent 的 MCP SSE 工具集成

示例 4：使用 Langchain 1.0 create_agent 的 MCP Stdio 工具集成

示例 1：`example_1_mcp_tool_stdio.py`（标准 I/O 通信）

示例 2：`example_1_mcp_tool_sse.py`（服务器发送事件通信）